package com.thinkgem.jeesite.common.im.message.server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.QueryStringDecoder;
import io.netty.handler.codec.http.multipart.Attribute;
import io.netty.handler.codec.http.multipart.DefaultHttpDataFactory;
import io.netty.handler.codec.http.multipart.HttpPostRequestDecoder;
import io.netty.handler.codec.http.multipart.InterfaceHttpData;
import io.netty.handler.codec.http.multipart.InterfaceHttpData.HttpDataType;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.GlobalEventExecutor;
import static io.netty.handler.codec.http.HttpHeaders.Names.*;
import static io.netty.handler.codec.http.HttpMethod.*;
import static io.netty.handler.codec.http.HttpResponseStatus.*;
import static io.netty.handler.codec.http.HttpVersion.*;

/**
 * 
 * @description:消息业务处理类
 * @fileName:IMServerHandler.java
 * @createTime:2015年7月23日 下午7:38:22
 * @author:肖震
 * @version 1.0.0
 * 
 */
public class IMServerHandler extends SimpleChannelInboundHandler<Object>{ 

	public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
	
	private static final String WEBSOCKET_PATH = "/websocket";
	
	// 在线用户状态map,由ip：在线状态组成
	private static final Map<String, String> onlineMap = new HashMap<String, String>();
	
	// 在线用户channel map,由ip：channel组成
	private static final Map<String, Channel> onlineUserChannelMap = new HashMap<String, Channel>();
	
	private WebSocketServerHandshaker handshaker;
	
	@Override
	public void handlerAdded(ChannelHandlerContext ctx)throws Exception{ // (2)
		Channel incoming = ctx.channel();
		for (Channel channel : channels){
			channel.writeAndFlush("[SERVER] - " + incoming.remoteAddress() + " 加入\n");
		}
		channels.add(ctx.channel());
	}
	
	@Override
	public void channelReadComplete(ChannelHandlerContext ctx){
		ctx.flush();
	}
	
	@Override
	public void handlerRemoved(ChannelHandlerContext ctx) throws Exception{
		Channel incoming = ctx.channel();
		for (Channel channel : channels){
			channel.writeAndFlush("[SERVER] - " + incoming.remoteAddress() + " 离开\n");
		}
		channels.remove(ctx.channel());
	}
	
	@Override
	protected void channelRead0(ChannelHandlerContext ctx, Object msg)throws Exception{
		// http请求
		if (msg instanceof FullHttpRequest){
			handleHttpRequest(ctx, (FullHttpRequest) msg);
		}else if (msg instanceof WebSocketFrame){// websocket请求
			handleWebSocketFrame(ctx, (WebSocketFrame) msg);
		}
		
	}
	
	/**
	 * 
	 * @function: 获得本地webscoket地址
	 * @param req：http请求对象
	 * @return String ：返回地址字符串
	 * @exception
	 * @author:肖震
	 * @since 1.0.0
	 */
	private static String getWebSocketLocation(FullHttpRequest req){
		String location = req.headers().get(HOST) + WEBSOCKET_PATH;
		
		return "ws://" + location;
		
	}
	
	/**
	 * 
	 * @function: 处理websocket请求业务
	 * @param ctx
	 * @param frame
	 * @exception
	 * @author:肖震
	 * @since 1.0.0
	 */
	private void handleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame){
		// 是否关闭
		if (frame instanceof CloseWebSocketFrame){
			handshaker.close(ctx.channel(),(CloseWebSocketFrame) frame.retain());
			return;
		}
		// 是否ping
		if (frame instanceof PingWebSocketFrame){
			ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
			return;
		}
		// 非文本
		if (!(frame instanceof TextWebSocketFrame)){
			throw new UnsupportedOperationException(String.format(
					"%s frame types not supported", frame.getClass().getName()));
		}
		
		// 接收到的消息
		String s = ((TextWebSocketFrame) frame).text();
		// 发送者的通道
		Channel incoming = ctx.channel();
		System.out.println("收到客户[" + incoming.remoteAddress() + "]消息：" + s);
		// 从发送消息解析出接收人的用户id
		String acceptUserId = null;
		// 解析出接收者的用户id
		if (s != null && s.contains("||-||")){
			acceptUserId = s.split("||-||")[1];
		}
		// 点对点发送消息
		if (acceptUserId != null){
			Channel channel = onlineUserChannelMap.get(acceptUserId);
			if (channel != null){
				channel.writeAndFlush(new TextWebSocketFrame("["+ channel.remoteAddress() + "]:" + s + "\n"));
			}
		}else{		// 群发消息
			for (Channel channel : channels){
				if (channel != incoming){
					channel.writeAndFlush(new TextWebSocketFrame("[" + channel.remoteAddress() + "]:" + s + "\n"));
				}else{
					channel.writeAndFlush(new TextWebSocketFrame("[" + incoming.remoteAddress() + "]:" + s + "\n"));
				}
			}
		}
	}
	
	/**
	 * 
	 * @function: http请求处理
	 * @param ctx
	 * @param req
	 * @throws Exception
	 * @exception
	 * @author:肖震
	 * @since 1.0.0
	 */
	private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception{
		System.out.println("handleHttpRequest method=========="+ req.getMethod());
		System.out.println("handleHttpRequest uri==========" + req.getUri());
		// http get请求
		if (req.getMethod() == GET){
			httpGetDeal(ctx, req);
		}
		if (req.getMethod() == POST){
			httpPostDeal(ctx, req);
			return;
		}
	}
	
	/**
	 * @function: http post方式处理
	 * @param ctx
	 * @param req
	 * @throws IOException
	 * @exception
	 * @author:肖震
	 * @since 1.0.0
	 */
	private void httpPostDeal(ChannelHandlerContext ctx, FullHttpRequest req)throws IOException{
		// 处理POST请求
		HttpPostRequestDecoder decoder = new HttpPostRequestDecoder(new DefaultHttpDataFactory(false), req);
		InterfaceHttpData postData = decoder.getBodyHttpData("q");
																   // 读取从客户端传过来的参数
		String question = "";
		if (postData.getHttpDataType() == HttpDataType.Attribute){
			Attribute attribute = (Attribute) postData;
			question = attribute.getValue();
			System.out.println("q:" + question);
		}
		if (question != null && !question.equals("")){
			String data = question;
			FullHttpResponse res = new DefaultFullHttpResponse(HTTP_1_1, OK,
					Unpooled.wrappedBuffer(data.getBytes(CharsetUtil.UTF_8)));
			
			res.headers().set(CONTENT_TYPE, "text/html; charset=UTF-8");
			res.headers().set(ACCESS_CONTROL_ALLOW_ORIGIN, "*");
			res.headers().set(CONTENT_LENGTH, res.content().readableBytes());
			sendHttpResponse(ctx, req, res);
		}
		return;
	}
	
	/**
	 * @function: http get方式处理
	 * @param ctx
	 * @param req void
	 * @exception
	 * @author:肖震
	 * @since 1.0.0
	 */
	private void httpGetDeal(ChannelHandlerContext ctx, FullHttpRequest req)
	{
		// 如果get请求地址中包括websocket
		if (req.getUri().equals("/websocket")){
			// Handshake
			WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
					getWebSocketLocation(req), null, false);
			handshaker = wsFactory.newHandshaker(req);
			if (handshaker == null){
				wsFactory.sendUnsupportedVersionResponse(ctx.channel());
			}
			else
			{
				
				handshaker.handshake(ctx.channel(), req);
				
			}
		}
		else
		// 处理get请求
		{
			QueryStringDecoder decoder = new QueryStringDecoder(req.getUri());
			Map<String, List<String>> parame = decoder.parameters();
			// 读取从客户端传过来的参数
			List<String> q = parame.get("q");
			// 参数值
			String question = q.get(0);
			// 根据参数值处理对应的业务
			if (question != null && question.equals("getData"))
			{
				System.out.println("r :" + question);
				String data = question;
				// 返回内容
				FullHttpResponse res = new DefaultFullHttpResponse(HTTP_1_1,
						OK, Unpooled.wrappedBuffer(data
								.getBytes(CharsetUtil.UTF_8)));
				
				res.headers().set(CONTENT_TYPE, "text/html; charset=UTF-8");
				res.headers().set(ACCESS_CONTROL_ALLOW_ORIGIN, "*");
				res.headers()
						.set(CONTENT_LENGTH, res.content().readableBytes());
				sendHttpResponse(ctx, req, res);
			}
			// 根据参数值处理对应的业务
			if (question != null && question.equals("getClients"))
			{
				System.out.println("r :" + question);
				String data = getOnlineClients();
				// 返回内容
				FullHttpResponse res = new DefaultFullHttpResponse(HTTP_1_1,
						OK, Unpooled.wrappedBuffer(data
								.getBytes(CharsetUtil.UTF_8)));
				
				res.headers().set(CONTENT_TYPE, "text/html; charset=UTF-8");
				res.headers().set(ACCESS_CONTROL_ALLOW_ORIGIN, "*");
				res.headers()
						.set(CONTENT_LENGTH, res.content().readableBytes());
				sendHttpResponse(ctx, req, res);
			}
		}
	}
	
	/**
	 * 
	 * @function: 获得在线用户，并拼成字符串形式
	 * @return String
	 * @exception
	 * @author:肖震
	 * @since 1.0.0
	 */
	private static String getOnlineClients()
	{
		String clients = "无客服在线";
		StringBuffer sb = new StringBuffer();
		if (!onlineMap.isEmpty())
		{
			for (String key : onlineMap.keySet())
			{
				if (key != null)
				{
					sb.append(key).append("[").append(onlineMap.get(key))
							.append("]\r\n");
				}
			}
		}
		if (sb.length() > 0)
		{
			clients = sb.toString();
		}
		System.out.println("客户：" + clients);
		return clients;
	}
	
	/**
	 * 
	 * @function: http请求响应处理
	 * @param ctx
	 * @param req
	 * @param res void
	 * @exception
	 * @author:肖震
	 * @since 1.0.0
	 */
	private static void sendHttpResponse(ChannelHandlerContext ctx,
			FullHttpRequest req, FullHttpResponse res)
	{
		// Generate an error page if response getStatus code is not OK (200).
		if (res.getStatus().code() != 200)
		{
			ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(),
					CharsetUtil.UTF_8);
			res.content().writeBytes(buf);
			buf.release();
			HttpHeaders.setContentLength(res, res.content().readableBytes());
		}
		
		// Send the response and close the connection if necessary.
		ChannelFuture f = ctx.channel().writeAndFlush(res);
		if (!HttpHeaders.isKeepAlive(req) || res.getStatus().code() != 200)
		{
			f.addListener(ChannelFutureListener.CLOSE);
		}
	}
	
	@Override
	public void channelActive(ChannelHandlerContext ctx)
		throws Exception
	{ // (5)
		Channel incoming = ctx.channel();
		onlineMap.put(((InetSocketAddress) incoming.remoteAddress())
				.getAddress().getHostAddress().toString(), "在线");
		onlineUserChannelMap.put(((InetSocketAddress) incoming.remoteAddress())
				.getAddress().getHostAddress().toString(), incoming);
		System.out.println("imClient[" + incoming.remoteAddress() + "]在线");
	}
	
	@Override
	public void channelInactive(ChannelHandlerContext ctx)
		throws Exception
	{ // (6)
		Channel incoming = ctx.channel();
		onlineMap.put(((InetSocketAddress) incoming.remoteAddress())
				.getAddress().getHostAddress().toString(), "掉线");
		onlineUserChannelMap.remove(((InetSocketAddress) incoming
				.remoteAddress()).getAddress().getHostAddress().toString());
		
		System.out.println("imClient[" + incoming.remoteAddress() + "]掉线");
	}
	
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
	{
		Channel incoming = ctx.channel();
		System.out.println("imClient[" + incoming.remoteAddress() + "]异常");
		// 当出现异常就关闭连接
		cause.printStackTrace();
		ctx.close();
	}
}
